package com.rem.flink;

import cn.hutool.json.JSONUtil;
import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.Set;

/**
 * @author Rem
 * @date 2022-09-30
 */

public class KafkaToRedis2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("10.224.18.157:9092,10.224.18.158:9092,10.224.18.159:9092")
                .setTopics("clicks")
                .setGroupId("consumer-group")
                .setStartingOffsets(OffsetsInitializer.committedOffsets())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStreamSource<Event> eventSource = env.addSource(new ClickSource());


        SingleOutputStreamOperator<Event> count = eventSource.assignTimestampsAndWatermarks(
                //插入水位线 针对乱序设置延迟时间  顺序不需要
                WatermarkStrategy.<Event>forMonotonousTimestamps()
                        //提取时间戳
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()));

        count.keyBy(Event::getUrl)
                //设置窗口 滚动窗口（tumbling window）、滑动窗口（sliding window）和会话窗口（session window）。
                .window(SlidingEventTimeWindows.of(Time.seconds(15), Time.seconds(5)))
                .aggregate(new UrlViewCountAgg(), new UrlViewCountResult())
                .print();

        env.execute();


        FlinkJedisSentinelConfig conf = new FlinkJedisSentinelConfig.Builder()
                //   .setSentinels(Set.of("10.224.18.157:26379", "10.224.18.158:26379", "10.224.18.159:26379"))
                .setSentinels(Set.of("localhost:6380", "localhost:6381", "localhost:6382"))
//                .setMasterName("RedisMaster")
                .setMasterName("mymaster")
//                .setPassword("Aldi2020!@#")
                .setPassword("123456")
                .build();

        eventSource.print();
        // eventSource.addSink(new RedisSink<Event>(conf, new MyRedisMapper()));


    }

    public static class MyRedisMapper implements RedisMapper<Event> {

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "clicks");
        }

        @Override
        public String getKeyFromData(Event event) {
            return event.getUser();
        }

        @Override
        public String getValueFromData(Event event) {
            return JSONUtil.toJsonStr(event);
        }
    }

    private static class UrlViewCountAgg implements AggregateFunction<Event, Long, Long> {
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(Event event, Long accumulator) {
            return accumulator + 1;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return null;
        }
    }

    private static class UrlViewCountResult extends ProcessWindowFunction<Long, Tuple4<String, String, Timestamp, Timestamp>, String, TimeWindow> {

        @Override
        public void process(String s, ProcessWindowFunction<Long, Tuple4<String, String, Timestamp, Timestamp>, String, TimeWindow>.Context context,
                            Iterable<Long> elements, Collector<Tuple4<String, String, Timestamp, Timestamp>> out) throws Exception {

            long start = context.window().getStart();
            long end = context.window().getEnd();
            Long next = elements.iterator().next();
            out.collect(Tuple4.of("url:" + s, "时间段内访问频次："+elements.iterator().next(), new Timestamp(start), new Timestamp(end)));
        }


    }


}
